package com.test.cn.task4;

import jdk.nashorn.internal.runtime.ListAdapter;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName ListProcessFunction
 * @Description TODO
 * @Author huangjh5
 * @Date 2021/5/14 20:07
 */
public class ListProcessFunction extends ProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>> implements CheckpointedFunction {

    private List<Tuple2<String,Long>> bufferedElements;

    private ListState<Tuple2<String, Long>> operatorState;


    public ListProcessFunction() {
        bufferedElements = new ArrayList<>();
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println(" open ..");
    }

    @Override
    public void processElement(Tuple2<String, Long> value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        System.out.println("processElement... ");
        bufferedElements.add(value);
        if(bufferedElements.size() == 3) {
            //
            for(Tuple2<String,Long> element : bufferedElements) {
               out.collect(element);
            }
            bufferedElements.clear();
        }


    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("....snapshotState");
        this.operatorState.clear();
        for (Tuple2<String,Long> element : bufferedElements) {
            operatorState.add(element);
        }
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("....initializeState");
        //做出一个State
        ListStateDescriptor<Tuple2<String, Long>> operatarDemoDescriptor = new ListStateDescriptor<>(
                "operatarDemo",
                TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {
                })
        );

        operatorState = context.getOperatorStateStore().getListState(operatarDemoDescriptor);

        if (context.isRestored()) {
            for (Tuple2<String,Long> element: operatorState.get()) {
                bufferedElements.add(element);
            }
            System.out.println("....context.isRestored():true" + bufferedElements);

        }


    }
}
